package run;

import com.SpringBootApps;
import io.rsocket.Payload;
import io.rsocket.SocketAcceptor;
import io.rsocket.util.DefaultPayload;
import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.context.support.GenericApplicationContext;
import org.springframework.messaging.handler.annotation.MessageMapping;
import org.springframework.messaging.rsocket.RSocketRequester;
import org.springframework.messaging.rsocket.RSocketStrategies;
import org.springframework.messaging.rsocket.annotation.support.RSocketMessageHandler;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.Sinks;
import utils.Utils;

import java.time.Duration;
import java.util.logging.Level;

@Slf4j
@SpringBootTest(classes = {StartUpForRSocketClient.Dummy.class},
        webEnvironment = SpringBootTest.WebEnvironment.DEFINED_PORT)
public class StartUpForRSocketClient {
    Object lock = new Object();
    @Autowired
    GenericApplicationContext context;


    @Test
    public void verify() throws InterruptedException {


        Utils.print(context);
        synchronized (lock) {
            lock.wait();
        }
    }

    private static class ClientHandler {

        final Sinks.Many<String> fireForgetPayloads = Sinks.many().replay().all();

        @MessageMapping("receive")
        void receive(String payload) {
            this.fireForgetPayloads.emitNext(payload, Sinks.EmitFailureHandler.FAIL_FAST);
        }

        @MessageMapping("echo")
        String echo(String payload) {
            return payload;
        }

        @MessageMapping("echo-async")
        Mono<String> echoAsync(String payload) {
            return Mono.delay(Duration.ofMillis(10)).map(aLong -> payload + " async");
        }

        @MessageMapping("echo-stream")
        Flux<String> echoStream(String payload) {
            return Flux.interval(Duration.ofMillis(10)).map(aLong -> payload + " " + aLong).onBackpressureBuffer();
        }

        @MessageMapping("echo-channel")
        Flux<String> echoChannel(Flux<String> payloads) {
            return payloads.delayElements(Duration.ofMillis(10)).map(payload -> payload + " async");
        }
    }

    @Test
    public void request(){

        RSocketStrategies strategies = context.getBean(RSocketStrategies.class);
        SocketAcceptor responder = RSocketMessageHandler.responder(strategies, new ClientHandler());

        RSocketRequester requester = null;
        try {
            requester = RSocketRequester.builder()
                    .setupRoute("echo")
                    .rsocketStrategies(strategies)
                    .rsocketConnector(connector -> connector.acceptor(responder))
                    .tcp("localhost", 9889);

            // Trigger connection establishment.
            requester.rsocketClient().source().block();
            Thread.sleep(10000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            if (requester != null) {
                requester.rsocketClient().dispose();
            }
        }


    }



    @Import(value = SpringBootApps.class)
    @Configuration
    public static class Dummy {
    }

}
